package com.bw.wjw.function;


import com.bw.wjw.DateTimeUtil;
import com.bw.wjw.bean.PageViewBean;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 创建窗口函数实例，对每分钟窗口中数据进行计算
 * @author xuanyu
 * @date 2025/1/9
 */
public class PageViewWindowFunction
        implements WindowFunction<PageViewBean, String, String, TimeWindow> {

    @Override
    public void apply(String key,
                      TimeWindow window,
                      Iterable<PageViewBean> input,
                      Collector<String> out) throws Exception {
        // key：表示分组字段；window：表示时间窗口，可以获取窗口开始和结束时间；input：表示窗口中所有数据
        String windowStart = DateTimeUtil.convertLongToString(window.getStart(), "yyyy-MM-dd HH:mm:ss");
        String windowEnd = DateTimeUtil.convertLongToString(window.getEnd(), "yyyy-MM-dd HH:mm:ss");

        // 遍历窗口中数据
        long pvCount = 0L;
        long pvDuringTime = 0L;
        long uvCount = 0L;
        long sessionCount = 0L;
        for (PageViewBean pageViewBean : input) {
            pvCount += pageViewBean.getPvCount();
            pvDuringTime += pageViewBean.getPvDuringTime();
            uvCount += pageViewBean.getUvCount();
            sessionCount += pageViewBean.getSessionCount();
        }

        // 输出
        String output = windowStart + "," + windowEnd + "," + key
                + "," + pvCount + "," + pvDuringTime + "," + uvCount + "," + sessionCount
                + "," + System.currentTimeMillis();
        out.collect(output);
    }

}
